/*
 * File: VMSEncoding.java Author: Fating <xingon@163.com>
 * 
 * Created on 2014年10月29日上午9:34:48
 * 
 * Copyright (c) 2014, www.vip.com All Rights Reserved.
 */
package com.tc.vms.kafka;

import com.google.protobuf.ByteString;
import com.tc.vms.MessageWrap;
import com.tc.vms.encoding.Message.KeyPair;
import com.tc.vms.encoding.Message.VMSMessage;
import com.tc.vms.encoding.Message.VMSMessage.Builder;
import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;

import java.util.Map.Entry;

public class KafkaMessageSerializer implements Encoder<MessageWrap> {

    public KafkaMessageSerializer(VerifiableProperties properties) {
    }

    @Override
    public byte[] toBytes(MessageWrap message) {

        Builder builder = VMSMessage.newBuilder().setPayload(
                ByteString.copyFrom(message.getPayload()));

        builder.setTtl(message.getTtl());
        builder.setWeight(message.getPriority());

        if (message.getMessageId() != null) {
            builder.setMessageId(message.getMessageId());
        }

        for (Entry<String, Object> entry : message.getProperties()) {
            KeyPair kp = KeyPair.newBuilder().setKey(entry.getKey())
                    .setValue(String.valueOf(entry.getValue())).build();
            builder.addProperties(kp);
        }

        for (String routing : message.getRoutingKeys()) {
            builder.addRoutingKeys(routing);
        }
        return builder.build().toByteArray();
    }

}
